package ttl

import (
	"fmt"
	"github.com/sirupsen/logrus"
	"strconv"
	"strings"
	"time"

	redis "github.com/go-redis/redis/v7"
	v1 "qabx.net/common/types/v1"
	qianxinutilsclientv2 "qabx.net/common/utils/v2/client"
)

var log *logrus.Logger

func SetLogger(logger *logrus.Logger) {
	log = logger
}

type TTLOperator interface {
	IncrAndExpire(key string, timeout time.Duration) (int64, error)
	Expire(key string, timeout time.Duration) error
	Exist(key string) bool
	Set(key string, timeout time.Duration) error
	SetV(key string, value string, timeout time.Duration) error
	Incr(key string) (int64, error)
	IncrBy(key string, value int64, timeout time.Duration) error
	QueryV(key string) string
	GenerateKey(key ...string) string
	InitClient(string, string) error
	newExpiredEventMonitor(do func(string)) error
	SetList(key string, value string) error
	RangeList(key string) (error, []string)
	DelList(key string) error
	Hset(key string, field string, val string) bool
	Hget(key string, field string) (bool, string)
	Scan(pattern string, count int64) []string
	GetNotifyChan() chan string
	Close()
	HLen(key string) (int64, error)
}

type RedisTTLOperator struct {
	redisClient *v1.RedisConnection
	//redisClient *v1.RedisClusterConnection
	pubsub     *redis.PubSub
	stop       chan struct{}
	stoped     chan struct{}
	rebuild    int
	NotifyChan chan string
}

func (r *RedisTTLOperator) GetNotifyChan() chan string {
	return r.NotifyChan
}

func (r *RedisTTLOperator) do(key string) {

	// log.Message("INFO", "%s has been expired\n", key)
	// resultKey := "result-" + key
	// r.DelList(resultKey)
	//log.Warnf("%s has been deleted\n", key)
	r.NotifyChan <- key
}

func (r *RedisTTLOperator) Close() {
	r.stop <- struct{}{}
	<-r.stoped
	r.redisClient.Client.Close()
}

func (r *RedisTTLOperator) newExpiredEventMonitor(callback func(string)) error {
	r.pubsub = r.redisClient.Client.PSubscribe("__keyevent@0__:expired")
	msgChan := make(chan redis.Message, 1)
	go func() {
		for {

			go func() {
				msg, err := r.pubsub.ReceiveMessage()
				if err != nil {
					if !strings.Contains(err.Error(), "client is closed") {
						log.Errorf(err.Error())
					}
					msgChan <- redis.Message{}
					return
				}
				msgChan <- *msg
			}()

			select {
			case <-r.stop:
				goto end
			case msg := <-msgChan:
				log.Debugf("Redis recv message: %s", msg.String())
				if msg.Payload != "" {
					callback(msg.Payload)
				}

			}
			time.Sleep(time.Second)

		}
	end:
		r.pubsub.Close()
		r.stoped <- struct{}{}
	}()
	return nil
}

func (r *RedisTTLOperator) ZGet(key string, startTime int64, endTime int64) (res []redis.Z) {
	op := &redis.ZRangeBy{
		Min: strconv.FormatInt(startTime, 10),
		Max: strconv.FormatInt(endTime, 10),
	}
	res, err := r.redisClient.Client.ZRangeByScoreWithScores(key, op).Result()
	if err != nil {
		log.Errorf("ZRangeByScoreWithScores redis %s failed: %s", key, startTime, endTime, err.Error())
		return
	}
	return

}

func (r *RedisTTLOperator) Scan(pattern string, count int64) []string {
	scanCMD := r.redisClient.Client.Scan(0, pattern, count)
	s, _, err := scanCMD.Result()
	if err != nil {
		log.Errorf("scan redis %s failed: %s", pattern, err.Error())
		return []string{}
	}

	return s

}

func (r *RedisTTLOperator) LPushList(key string, val interface{}) (res int64, err error) {
	res, err = r.redisClient.Client.LPush(key, val).Result()
	return
}

func (r *RedisTTLOperator) RPopList(key string) (str string, err error) {
	str, err = r.redisClient.Client.RPop(key).Result()
	return
}

func (r *RedisTTLOperator) InitClient(url string, authStr string, redisDB int) error {
	var err error
	r.stop = make(chan struct{})
	r.stoped = make(chan struct{})
	r.NotifyChan = make(chan string)
	//if r.redisClient = qianxinutilsclientv2.NewRedisClusterConnectionWithParams([]string{url}, 102400, authStr); r.redisClient.Client == nil {
	if r.redisClient = qianxinutilsclientv2.NewRedisConnectionWithParams(url, 102400, redisDB, authStr); r.redisClient.Client == nil {
		err = fmt.Errorf("init redis failed")
		log.Errorf(err.Error())
		return err
	} else {
		cmdOnce := r.redisClient.Client.Ping()
		if err = cmdOnce.Err(); err != nil {
			panic(err.Error())
			log.Errorf("init redis failed: %s", err.Error())
			return err
		}

		go func() {
			var cmd *redis.StatusCmd
			for {
				cmd = r.redisClient.Client.Ping()
				if err = cmd.Err(); err != nil {
					log.Errorf("init redis failed: %s", err.Error())
					r.rebuild = 1
					for {
						//if r.redisClient = qianxinutilsclientv2.NewRedisClusterConnectionWithParams([]string{url}, 102400, authStr); r.redisClient.Client == nil {
						if r.redisClient = qianxinutilsclientv2.NewRedisConnectionWithParams(url, 102400, 0, authStr); r.redisClient.Client == nil {
							err = fmt.Errorf("rebuild redis client failed")
							log.Errorf(err.Error())
							time.Sleep(time.Second)
							continue
						}
						r.rebuild = 0
						break
					}
				}

				time.Sleep(3 * time.Second)
			}

		}()

	}
	r.newExpiredEventMonitor(r.do)
	return nil
}

func (r *RedisTTLOperator) Expire(key string, timeout time.Duration) error {
	boolCmd := r.redisClient.Client.Expire(key, timeout)
	return boolCmd.Err()
}

func (r *RedisTTLOperator) IncrAndExpire(key string, timeout time.Duration) (int64, error) {
	statusCmd := r.redisClient.Client.Incr(key)
	if statusCmd.Err() != nil {
		return statusCmd.Result()
	} else {
		return 1, r.Expire(key, timeout)
	}
}

func (r *RedisTTLOperator) Exist(key string) bool {
	statusCmd := r.redisClient.Client.Exists(key)
	result, err := statusCmd.Result()
	if err == nil && result == 1 {
		return true
	} else if err != nil {
		log.Errorf("Query redis exist error:%s", err.Error())
	}
	return false
}

func (r *RedisTTLOperator) Set(key string, timeout time.Duration) error {
	var statusCmd *redis.StatusCmd
	for i := 0; i < 10; i++ {
		statusCmd = r.redisClient.Client.Set(key, "1", timeout)
		if statusCmd.Err() != nil {
			if strings.Contains(statusCmd.Err().Error(), "client is closed") {
				log.Warnf("Set redis list key ttl failed: %s,retry", statusCmd.Err().Error())
				continue
			}
		}
		break
	}
	return statusCmd.Err()
}

func (r *RedisTTLOperator) SetV(key string, value string, timeout time.Duration) error {
	var statusCmd *redis.StatusCmd
	for i := 0; i < 10; i++ {
		statusCmd = r.redisClient.Client.Set(key, value, timeout)
		if statusCmd.Err() != nil {
			if strings.Contains(statusCmd.Err().Error(), "client is closed") {
				log.Warnf("Set redis list key ttl failed: %s,retry", statusCmd.Err().Error())
				continue
			}
		}
		break
	}
	return statusCmd.Err()
}

func (r *RedisTTLOperator) Incr(key string) (int64, error) {
	return r.redisClient.Client.Incr(key).Result()
}

func (r *RedisTTLOperator) IncrBy(key string, value int64) error {
	statusCmd := r.redisClient.Client.IncrBy(key, value)
	return statusCmd.Err()
}

func (r *RedisTTLOperator) SetList(key string, value string) error {
	// var
	var statusCmd *redis.IntCmd
	var boolCmd *redis.BoolCmd

	for i := 0; i < 10; i++ {
		statusCmd = r.redisClient.Client.LPush(key, value)
		boolCmd = r.redisClient.Client.Expire(key, time.Second*3600)
		if boolCmd.Err() != nil {
			if strings.Contains(boolCmd.Err().Error(), "client is closed") {
				log.Warnf("Set redis list key ttl failed: %s", boolCmd.Err().Error())
				continue
			} else {
				log.Warnf("Set redis list key ttl failed: %s", boolCmd.Err().Error())
				return statusCmd.Err()
			}
		}
		break
	}
	return statusCmd.Err()

}

func (r *RedisTTLOperator) RangeList(key string) (error, []string) {
	res := make([]string, 5)
	var stringSliceCmd *redis.StringSliceCmd
	var err error
	for i := 0; i < 10; i++ {
		stringSliceCmd = r.redisClient.Client.LRange(key, 0, -1)
		err = stringSliceCmd.Err()
		if err != nil && !strings.Contains(err.Error(), "MOVED") {
			if strings.Contains(stringSliceCmd.Err().Error(), "client is closed") {
				continue
			}
			log.Errorf("Range redis list key failed: %s", err.Error())
			return err, res
		}
		break
	}

	res, err = stringSliceCmd.Result()
	if err != nil {
		log.Errorf("Get redis list key res failed: %s", err.Error())
		return err, res
	}
	return nil, res
}

func (r *RedisTTLOperator) DelList(key string) error {
	var intCmd *redis.IntCmd
	var err error
	for i := 0; i < 10; i++ {
		intCmd = r.redisClient.Client.Del(key)
		err = intCmd.Err()
		if err != nil {
			if strings.Contains(intCmd.Err().Error(), "client is closed") {
				continue
			}
			// if strings.Contain(err.Error(), "client is closed") {
			// 	log.Message("ERROR", "Redis client closed, nead restart")
			// }

			log.Errorf("Delete redis list key failed: %s", err.Error())
			return err
		}
		break
	}

	_, err = intCmd.Result()
	if err != nil {
		log.Errorf("delete redis list key failed: %s", err.Error())
		return err
	}
	// if res != 1 {
	// 	err = fmt.Errorf("Delete redis list key failed")
	// 	log.Message("ERROR", err.Error())
	// 	return err
	// }
	return nil
}

func (r *RedisTTLOperator) Query(key string) string {
	res := ""
	var durationCmd *redis.DurationCmd
	var err error

	for i := 0; i < 10; i++ {
		durationCmd = r.redisClient.Client.TTL(key)
		err = durationCmd.Err()
		if err != nil {
			if strings.Contains(durationCmd.Err().Error(), "client is closed") {
				continue
			}
			log.Errorf("get redis key %s ttl failed: %s", key, err.Error())
			return ""
		}
		break
	}

	ttl, err := durationCmd.Result()
	if err != nil {
		log.Errorf("Get redis key failed: %s", err.Error())
		return res
	}

	if ttl < 0 {
		if -1 == ttl.Seconds() {
			log.Errorf("The key will not expire.")
		} else if -2 == ttl.Seconds() {
			log.Warnf("The key %s does not exist.", key)
		} else {
			log.Errorf("Get %s ttl unexpected error.", key)
		}
	} else {
		res = strconv.Itoa(int(ttl.Seconds()))
	}

	return res
}

func (r *RedisTTLOperator) QueryV(key string) string {
	var stringCmd *redis.StringCmd
	var err error

	for i := 0; i < 10; i++ {
		stringCmd = r.redisClient.Client.Get(key)
		err = stringCmd.Err()
		if err != nil {
			if strings.Contains(stringCmd.Err().Error(), "client is closed") {
				continue
			}
			log.Errorf("Get redis key %s failed: %s", key, err.Error())
			return ""
		}
		break
	}

	res, err := stringCmd.Result()
	if err != nil {
		log.Errorf("Get redis key %s failed: %s", key, err.Error())
		return ""
	}

	return res
}

func (r *RedisTTLOperator) Hset(key string, field string, val string, timeout ...time.Duration) bool {
	var boolCmd *redis.BoolCmd
	var err error
	for i := 0; i < 10; i++ {
		boolCmd = r.redisClient.Client.HMSet(key, field, val)
		err = boolCmd.Err()
		if err != nil {
			if strings.Contains(boolCmd.Err().Error(), "client is closed") {
				continue
			}
			log.Errorf("Set redis hash cmd key %s failed: %s", key, err.Error())
			return false
		}
		break
	}

	res, err := boolCmd.Result()
	if err != nil {
		log.Errorf("Set redis hash result key %s failed: %s", key, err.Error())
		return false
	}

	if len(timeout) > 0 {
		err = r.Expire(key, timeout[0])
		if err != nil {
			return false
		}
	}

	return res
}

func (r *RedisTTLOperator) Hget(key string, field string) (bool, string) {
	var stringCmd *redis.StringCmd
	var err error

	for i := 0; i < 10; i++ {
		stringCmd = r.redisClient.Client.HGet(key, field)
		err = stringCmd.Err()
		if err != nil {
			if strings.Contains(stringCmd.Err().Error(), "client is closed") {
				continue
			}
			log.Errorf("Get redis hash cmd key %s failed: %s", key, err.Error())
			return false, ""
		}
		break
	}
	// if err != nil {
	// 	log.Message("ERROR", "Get redis hash cmd key %s failed: %s", key, err.Error())
	// 	return false, ""
	// }

	res, err := stringCmd.Result()
	if err != nil {
		log.Errorf("Get redis hash result key %s failed: %s", key, err.Error())
		return false, res
	}

	return true, res
}

func (r *RedisTTLOperator) HKeys(key string) []string {
	keys := r.redisClient.Client.HKeys(key)
	result, err := keys.Result()
	if err != nil {
		return nil
	}
	return result
}

func (r *RedisTTLOperator) HGetAll(key string) (result map[string]string, err error) {
	result, err = r.redisClient.Client.HGetAll(key).Result()
	return result, err
}

func (r *RedisTTLOperator) HLen(key string) (result int64, err error) {
	result, err = r.redisClient.Client.HLen(key).Result()
	return result, err
}

func (r *RedisTTLOperator) HDel(key string, field string) error {
	boolCmd := r.redisClient.Client.HDel(key, field)
	return boolCmd.Err()
}

func (r *RedisTTLOperator) GenerateKey(key ...string) string {

	return strings.Join(key, "-")
}

func (r *RedisTTLOperator) Del(key string) error {
	boolCmd := r.redisClient.Client.Del(key)
	return boolCmd.Err()
}

func (r *RedisTTLOperator) QueryNum(key string) int {
	IntCmd := r.redisClient.Client.Get(key)
	err := IntCmd.Err()
	//log.Errorf("%s", err.Error())
	if err != nil && !strings.Contains(err.Error(), "MOVED") {
		log.Errorf("Get redis key %s failed: %s", key, err.Error())
		return 0
	}

	res, err := IntCmd.Int()
	if err != nil {
		log.Errorf("Get redis key %s failed: %s", key, err.Error())
		return 0
	}

	return res
}

func (r *RedisTTLOperator) Sadd(key string, val string, timeout ...time.Duration) error {
	sAddCmd := r.redisClient.Client.SAdd(key, val)

	if sAddCmd.Err() != nil {
		return sAddCmd.Err()
	}

	if len(timeout) > 0 {
		boolCmd := r.redisClient.Client.Expire(key, timeout[0])
		if boolCmd.Err() != nil {
			return boolCmd.Err()
		}
	}

	return nil
}

func (r *RedisTTLOperator) SMem(key string) (result []string, err error) {
	sMemCmd := r.redisClient.Client.SMembers(key)
	if sMemCmd.Err() != nil {
		return
	}
	result, err = sMemCmd.Result()
	if err != nil {
		return
	}
	return
}

func (r *RedisTTLOperator) SRem(key string, itemKey string) error {
	sAddCmd := r.redisClient.Client.SRem(key, itemKey)

	if sAddCmd.Err() != nil {
		return sAddCmd.Err()
	}
	return nil
}

func (r *RedisTTLOperator) SIsMember(setKey string, key string) bool {
	statusCmd := r.redisClient.Client.SIsMember(setKey, key)
	result, err := statusCmd.Result()
	if err != nil {
		log.Errorf("Query redis SIsMember error:%s", err.Error())
	}
	return result
}

func (r *RedisTTLOperator) SCARD(setKey string) int64 {
	scardCmd := r.redisClient.Client.SCard(setKey)
	result, err := scardCmd.Result()
	if err != nil {
		log.Errorf("get redis Scard error:%s", err.Error())
	}
	return result
}

func (r *RedisTTLOperator) Get(key string) (string, error) {
	cmd := r.redisClient.Client.Get(key)
	return cmd.Result()
}

func (r *RedisTTLOperator) ZAdd(key string, score float64, member string) error {
	z := &redis.Z{
		Score:  score,
		Member: member,
	}
	statusCmd := r.redisClient.Client.ZAdd(key, z)
	if statusCmd.Err() != nil {
		return statusCmd.Err()
	}
	return nil
}

func (r *RedisTTLOperator) ZIncr(key string, score float64, member string) error {
	z := &redis.Z{
		Score:  score,
		Member: member,
	}
	statusCmd := r.redisClient.Client.ZIncr(key, z)
	if statusCmd.Err() != nil {
		return statusCmd.Err()
	}
	return nil
}

func (r *RedisTTLOperator) ZRem(key string, member interface{}) error {
	cmd := r.redisClient.Client.ZRem(key, member)
	if cmd.Err() != nil {
		return cmd.Err()
	}
	return nil
}

func (r *RedisTTLOperator) ZRank(setKey string, key string) int64 {
	statusCmd := r.redisClient.Client.ZRank(setKey, key)
	result, err := statusCmd.Result()
	if err != nil {
		log.Errorf("Query redis ZRank error:%s", err.Error())
		return -1
	}
	return result
}

func (r *RedisTTLOperator) ZCount(key, min, max string) int64 {
	ZcountCmd := r.redisClient.Client.ZCount(key, min, max)
	result, err := ZcountCmd.Result()
	if err != nil {
		log.Errorf("Do redis Zcount error:%s", err.Error())
	}
	return result
}

func (r *RedisTTLOperator) ZRevRangeByScore(key string, min int32, max int32, limit int32) ([]string, error) {
	if max < 0 {
		max = 999999999
	}
	if limit < 0 {
		limit = 1
	}

	z := &redis.ZRangeBy{
		Min:   strconv.Itoa(int(min)),
		Max:   strconv.Itoa(int(max)),
		Count: int64(limit),
	}

	statusCmd := r.redisClient.Client.ZRevRangeByScore(key, z)
	if statusCmd.Err() != nil {
		return nil, statusCmd.Err()
	}
	return statusCmd.Result()
}

func (r *RedisTTLOperator) ZRevRange(key string, min int32, max int32) ([]string, error) {
	cmd := r.redisClient.Client.ZRevRange(key, int64(min), int64(max))
	return cmd.Result()
}

func (r *RedisTTLOperator) LPop(key string) (string, error) {
	cmd := r.redisClient.Client.LPop(key)
	return cmd.Result()
}

func (r *RedisTTLOperator) RPop(key string) (string, error) {
	cmd := r.redisClient.Client.RPop(key)
	return cmd.Result()
}

func (r *RedisTTLOperator) SPop(key string) (string, error) {
	cmd := r.redisClient.Client.SPop(key)
	return cmd.Result()
}

func (r *RedisTTLOperator) LPush(key string, value string) error {
	intCmd := r.redisClient.Client.LPush(key, value)

	_, err := intCmd.Result()
	if err != nil {
		log.Errorf("Set redis result key %s failed: %s", key, err.Error())
		return err
	}

	return nil
}

func (r *RedisTTLOperator) LRange(key string, start int64, end int64) ([]string, error) {
	intCmd := r.redisClient.Client.LRange(key, start, end)
	return intCmd.Result()
}

func (r *RedisTTLOperator) LLen(key string) (int64, error) {
	intCmd := r.redisClient.Client.LLen(key)
	return intCmd.Result()
}

func (r *RedisTTLOperator) LIndex(key string, index int64) (string, error) {
	result, err := r.redisClient.Client.LIndex(key, index).Result()
	if err != nil {
		return "", err
	}
	return result, nil
}
